package me.seawenc.datastash.inputs.udp;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.util.Arrays;
import java.util.Properties;
import java.util.function.Consumer;
import me.seawenc.datastash.DatastashApplication;
import me.seawenc.datastash.inputs.AbstractProducer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;

/**
 * udp消息消费者测试
 */
public class UdpProducer extends AbstractProducer {
    int port;
    private static DatagramSocket datagramSocket;
    public UdpProducer(Properties fileProps) {
        super(fileProps);
        port = Integer.parseInt(fileProps.getProperty("input.udp.port"));
        System.out.println("udp port："+port);
    }
    private synchronized DatagramSocket getSocket() throws SocketException {
        if(datagramSocket==null){
            datagramSocket=new DatagramSocket(this.port);
        }
        return datagramSocket;
    }

    @Override
    public void producer(Consumer<String> consumer) throws Exception {

        byte[] bytes = new byte[102400];
        while (DatastashApplication.isRunning){
            try {
                DatagramPacket datagramPacket = new DatagramPacket(bytes, 0, bytes.length);
                getSocket().receive(datagramPacket);
                String msg = new String(datagramPacket.getData(), 0, datagramPacket.getLength()).replace("\n","");
                send(consumer,msg);
            } catch (Exception e) {
                System.out.println("udp数据接收出错,error="+e.getMessage());
            }
        }
    }
}